﻿using System;
using System.Net;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using Adaptive.Agrona;
using Adaptive.Agrona.Concurrent;
using Microsoft.Extensions.Logging;
using Unity.Commons;
using Unity.Commons.Client;
using Unity.Commons.Client.Buffer;
using Unity.Commons.Client.Generator;
using Unity.Commons.Client.Statistics;
using Unity.Commons.Client.Threading;
using Unity.Connections;
using Unity.Credentials;
using Unity.Logger;
using Unity.Type;

namespace Unity.Kafka.Client
{
    public class UnityClient : IUnityClient, IUnityStatistics
    {
        private static readonly ILogger Logger = UnityLogger.ConsoleLoggerFactory.CreateLogger(nameof(IUnityClient));
        
        private static int InsufficientCapacity = -2;
        private static readonly int RESOURCE_THREAD_SLEEPING_PERIOD = 500_000_000;
        private static readonly int PREALLOCATED_LENGTH = 1024 * 1024 * 8;

        private UnityGenerator _generator;
        private readonly UnityGenerator _initGenerator;
        private readonly string _csiId;
        private readonly IUnityCredentials _credentials;
        private readonly IUnityBuffer _mainRingBuffer;
        private readonly IUnityBuffer _offlineRingBuffer;
        private readonly UnityBufferFile _offlineBufferFile;

        private readonly AgentRunner _agentRunner;
        private readonly IUnityResourceThread _resourceThread;

        private static readonly AtomicLong Generated = new AtomicLong();
        private static readonly AtomicLong Dropped = new AtomicLong();

        private static volatile bool IsBlocking = false;

        private readonly IAtomicBuffer PreAllocatedBuffer;
        private readonly int MaximumPreAllocatedSize;
        private int PreAllocatedPosition;
        private EntityType LastPreGeneratedEntityType = EntityType.Unknown;

        private bool _isDisposing;

        public UnityClient(string csiId, IUnityCredentials credentials, IUnityBuffer mainBuffer, IUnityBuffer offlineBuffer,
                           UnityBufferFile offlineBufferFile, IUnityProducer resourceProducer) 
            : this(csiId, credentials, mainBuffer, offlineBuffer, offlineBufferFile, resourceProducer, 0)
        { }
        
        UnityClient(string csiId, IUnityCredentials credentials, IUnityBuffer mainBuffer, IUnityBuffer offlineBuffer,
                    UnityBufferFile offlineBufferFile, IUnityProducer resourceProducer, int initialCounter)
        {
            _csiId = csiId;
            // Need verify csiId
            _credentials = credentials;
            _generator = new UnityGenerator(_csiId);
            _initGenerator = _generator;
            if(initialCounter > 0)
            {
                _generator.SetCounter(initialCounter);
            }
            _mainRingBuffer = mainBuffer;
            _offlineRingBuffer = offlineBuffer;
            _offlineBufferFile = offlineBufferFile;
            _mainRingBuffer.Start();
            _offlineRingBuffer?.Start();
            _resourceThread = new UnityResourceThread(this, resourceProducer, csiId);
            _agentRunner = new AgentRunner(new SleepingIdleStrategy(RESOURCE_THREAD_SLEEPING_PERIOD),
                _errorHandler, null, _resourceThread);
            AgentRunner.StartOnThread(_agentRunner);
            PreAllocatedBuffer = new UnsafeBuffer(new ByteBuffer(PREALLOCATED_LENGTH, 0));
            MaximumPreAllocatedSize = PREALLOCATED_LENGTH / UnityGenerator.MessageLength;
        }

        private long GetVersion()
        {
            return UnityTimestamp.GetNanoTime();
        }

        public bool Connect(int timeout)
        {
            Task<bool> connectTask = AsyncConnect();
            if (connectTask.Wait(timeout))
            {
                return connectTask.Result;
            }

            return false;
        }

        private Task<bool> AsyncConnect()
        {
            return _resourceThread.RunAsync(InternalConnect);
        }

        private bool InternalConnect()
        {
            _resourceThread.GetProducer().Connect(_credentials);
            return _mainRingBuffer.GetProducer().Connect(_credentials);
        }

        public void Close()
        {
            _mainRingBuffer.Close();
            _offlineRingBuffer?.Close();
            _offlineBufferFile?.Close();
            _resourceThread?.Close();
            _agentRunner?.Dispose();
        }

        public void Dispose()
        {
            Close();
        }

        private int InternalClaim(IUnityBuffer buffer)
        {
            int claim;
            
            do
            {
                claim = buffer.Claim(UnityGenerator.MessageLength);
            } while (IsBlocking && claim == InsufficientCapacity);
            return claim;
        }

        public int Generate(EntityType entityType, IMutableDirectBuffer buffer, int offset)
        {
            buffer.BoundsCheck(offset, UnityGenerator.MaximumIdLength + sizeof(long));
            long version = GetVersion();
            int idLength = 0;
            int mainClaim = InternalClaim(_mainRingBuffer);
            if(mainClaim > 0)
            {
                idLength = GenerateIdViaRingBuffer(entityType, version, _mainRingBuffer, mainClaim, buffer, offset);
            }
            if (idLength == 0 && _offlineRingBuffer != null)
            {
                int offlineClaim = _offlineRingBuffer.Claim(UnityGenerator.MessageLength);
                if(offlineClaim > 0)
                {
                    idLength = GenerateIdViaRingBuffer(entityType, version, _offlineRingBuffer, offlineClaim, buffer, offset);
                }   
            }
            if (idLength == 0)
            {
                idLength = _generator.GenerateId(buffer, offset);
                if(idLength == (int) UnityGenerator.CounterStatus.Overflow)
                {
                    idLength = RotateGeneratorAndGenerate(buffer, offset);
                }
                Generated.IncrementAndGet();
            }
            if(mainClaim == InsufficientCapacity)
            {
                Dropped.IncrementAndGet();
            }
            buffer.PutLong(offset + idLength, version);
            return idLength;
        }

        private int GenerateIdViaRingBuffer(EntityType entityType, long version, IUnityBuffer buffer, int claim, IMutableDirectBuffer userBuffer, int userOffset)
        {
            int idLength = 0;
            try
            {
                idLength = _generator.GenerateMetadataAndCopy(entityType, version, buffer.GetBuffer(), claim, userBuffer, userOffset);
                if(idLength == (int) UnityGenerator.CounterStatus.Overflow)
                {
                    idLength = RotateGeneratorAndGenerate(entityType, version, buffer, claim, userBuffer, userOffset);
                }
                Generated.IncrementAndGet();
                buffer.Commit(claim);
            }
            catch(Exception exception)
            {
                buffer.Abort(claim);
            }
            return idLength;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        private int RotateGeneratorAndGenerate(IMutableDirectBuffer buffer, int offset)
        {
            int idLength = _generator.GenerateId(buffer, offset);
            if(idLength == (int) UnityGenerator.CounterStatus.Overflow)
            {
                _generator = _resourceThread.GetNewGenerator();
                idLength = _generator.GenerateId(buffer, offset);
            }
            return idLength;
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        private int RotateGeneratorAndGenerate(EntityType entityType, long version, IUnityBuffer buffer, int claim, IMutableDirectBuffer userBuffer, int userOffset)
        {
            int idLength = _generator.GenerateMetadataAndCopy(entityType, version, buffer.GetBuffer(), claim, userBuffer, userOffset);
            if(idLength == (int) UnityGenerator.CounterStatus.Overflow)
            {
                _generator = _resourceThread.GetNewGenerator();
                idLength = _generator.GenerateMetadataAndCopy(entityType, version, buffer.GetBuffer(), claim, userBuffer, userOffset);
            }
            return idLength;
        }

        public string GetCsiId()
        {
            return _csiId;
        }

        public ClientMode GetMode()
        {
            return _mainRingBuffer.GetMode();
        }

        public IUnityStatistics GetStatistics()
        {
            return this;
        }

        public long Update(EntityType entityType, IMutableDirectBuffer buffer, int offset, int length)
        {
            if(buffer == null)
            {
                throw new ArgumentNullException(nameof(buffer));
            }
            long version = GetVersion();
            int mainClaim = InternalClaim(_mainRingBuffer);
            bool isUpdated = false;
            if(mainClaim > 0)
            {
                isUpdated = UpdateViaRingBuffer(buffer, offset, length, entityType, version, _mainRingBuffer, mainClaim);
            }

            if(!isUpdated && _offlineRingBuffer != null)
            {
                int offlineClaim = _offlineRingBuffer.Claim(UnityGenerator.MessageLength);
                UpdateViaRingBuffer(buffer, offset, length, entityType, version, _offlineRingBuffer, offlineClaim);
            }

            if (mainClaim == InsufficientCapacity)
            {
                Dropped.IncrementAndGet();
            }
            return version;
        }

        private bool UpdateViaRingBuffer(IMutableDirectBuffer buffer, int offset, int length, EntityType entityType, long version, IUnityBuffer ringBuffer, int claim)
        {
           try
           {
               _generator.UpdateMetadata(buffer, offset, length, entityType, version, ringBuffer.GetBuffer(), claim);
               Generated.IncrementAndGet();
               ringBuffer.Commit(claim);
               return true;
           }
           catch(Exception exception)
           {
               ringBuffer.Abort(claim);
               Logger.LogError(exception, "Unexpected error during update metadata, message will be aborted");
           }
           return false;
        }

        public long GetPublishedStream()
        {
            return _mainRingBuffer.GetPublished();
        }

        public long GetPublishedOffline()
        {
            return _offlineRingBuffer?.GetPublished() ?? 0;
        }

        public long GetGenerated()
        {
            return Generated.Get();
        }

        public long GetDropped()
        {
            return Dropped.Get() + _mainRingBuffer.GetDropped() 
                + (_offlineRingBuffer?.GetDropped() ?? 0);
        }

        public long GetBufferCapacity()
        {
            return _mainRingBuffer.GetBufferCapacity();
        }

        public long GetBufferSize()
        {
            return _mainRingBuffer.GetBufferSize();
        }

        public long GetOfflineBufferCapacity()
        {
            return _offlineRingBuffer?.GetBufferCapacity() ?? 0;
        }

        public long GetOfflineBufferSize()
        {
            return _offlineRingBuffer?.GetBufferSize() ?? 0;
        }
        
        readonly ErrorHandler _errorHandler = (Exception exception) =>
        {
            Logger.LogError("Internal Unity exception", exception);
        };
    }
}
